/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.lastfm;

import aiaudio.lastfm.hbase.SchemeCommunicator;
import de.umass.lastfm.CallException;
import de.umass.lastfm.PaginatedResult;
import de.umass.lastfm.User;
import java.io.IOException;
import java.util.Collection;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class UserNeighboursExtracter {

    private String startId;
    SchemeCommunicator from;
    SchemeCommunicator to;

    public UserNeighboursExtracter(SchemeCommunicator from, SchemeCommunicator to) {
        this.from = from;
        this.to = to;
    }

    public UserNeighboursExtracter(String startId, SchemeCommunicator from, SchemeCommunicator to) {
        this(from, to);
        this.startId = startId;
    }

    public void start() throws IOException {
        Scan scan = new Scan();
        if (startId != null && startId != "") {
            scan.setStartRow(Bytes.toBytes(startId));
        }
        HTable parentTable = from.getUsersTable();
        ResultScanner scanner = parentTable.getScanner(scan);

        Result res;
        while ((res = scanner.next()) != null) {
            byte[] userId = res.getRow();
            String uid = Bytes.toString(userId);
            System.out.println("processing: " + uid);
            getNeighbours(uid);
        }
    }

    private void getNeighbours(String uid) throws IOException {
        try {
            PaginatedResult<User> neighbours = User.getFriends(uid, LastFmDownloaderHBase.API_KEY);
            Collection<User> pageResults = neighbours.getPageResults();
            for (User user : pageResults) {
                to.insertUser(user);
                to.createLink(uid, user.getId());
            }
        } catch (CallException ex) {
            System.out.println("exception at user: " + uid);
        }
    }

    private void processUser(User user) throws IOException {
//        user = User.getInfo(user.getName(), LastFmDownloaderHBase.API_KEY);
//        String id = user.getId();
//        boolean exists1 = from.getUsersTable().exists(new Get(Bytes.toBytes(id)));
//        boolean exists2 = to.getUsersTable().exists(new Get(Bytes.toBytes(id)));
//
//        if(exists1 || exists2){
//            return;
//        }
    }
}
